package io.kiki.stack.netty.invoker;


import io.kiki.stack.netty.channelManager.Channel;
import io.kiki.stack.netty.channelManager.ChannelManager;
import io.kiki.stack.netty.channelManager.Url;
import io.kiki.stack.netty.command.Command;
import io.kiki.stack.netty.command.CommandFactory;
import io.kiki.stack.netty.exception.RemotingException;
import io.kiki.stack.netty.util.InetSocketAddressUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Setter
@Getter
public class ClientInvoker extends Invoker {

    public ClientInvoker(CommandFactory commandFactory, ChannelManager channelManager) {
        super(commandFactory, channelManager);
    }


    @Override
    public void oneway(Url url, Object request, RequestContext requestContext) throws RemotingException, InterruptedException {
        if (requestContext == null) {
            requestContext = new RequestContext();
        }
        final Channel channel = getChannelAndInitInvokeContext(url, requestContext);
        this.channelManager.check(channel);
        this.oneway(channel, request, requestContext);
    }


    @Override
    public Object invokeSync(Url url, Object request, RequestContext requestContext, int timeoutMillis) throws RemotingException, InterruptedException {
        // Use the specified timeout to overwrite the configuration in the url
        url.setConnectTimeout(timeoutMillis);
        if (requestContext == null) {
            requestContext = new RequestContext();
        }
        final Channel conn = getChannelAndInitInvokeContext(url, requestContext);
        this.channelManager.check(conn);
        return this.invokeSync(conn, request, requestContext, timeoutMillis);
    }

    @Override
    public RpcResponseFuture invokeWithFuture(Url url, Object request, RequestContext requestContext, int timeoutMillis) throws RemotingException, InterruptedException {
        url.setConnectTimeout(timeoutMillis);
        if (requestContext == null) {
            requestContext = new RequestContext();
        }
        final Channel conn = getChannelAndInitInvokeContext(url, requestContext);
        this.channelManager.check(conn);
        return this.invokeWithFuture(conn, request, requestContext, timeoutMillis);
    }


    @Override
    public void invokeWithCallback(Url url, Object request, RequestContext requestContext, Callback callback, int timeoutMillis) throws RemotingException, InterruptedException {
        url.setConnectTimeout(timeoutMillis);
        if (requestContext == null) {
            requestContext = new RequestContext();
        }
        final Channel conn = getChannelAndInitInvokeContext(url, requestContext);
        this.channelManager.check(conn);
        this.invokeWithCallback(conn, request, requestContext, callback, timeoutMillis);
    }


    @Override
    protected void preProcessInvokeContext(RequestContext requestContext, Command command, Channel channel) {
        if (null != requestContext) {
            requestContext.putIfAbsent(RequestContext.CLIENT_LOCAL_IP, InetSocketAddressUtil.parseLocalIp(channel.getChannel()));
            requestContext.putIfAbsent(RequestContext.CLIENT_LOCAL_PORT, InetSocketAddressUtil.parseLocalPort(channel.getChannel()));
            requestContext.putIfAbsent(RequestContext.CLIENT_REMOTE_IP, InetSocketAddressUtil.parseRemoteIp(channel.getChannel()));
            requestContext.putIfAbsent(RequestContext.CLIENT_REMOTE_PORT, InetSocketAddressUtil.parseRemotePort(channel.getChannel()));
            requestContext.putIfAbsent(RequestContext._INVOKE_REQUEST_ID, command.getId());
        }
    }

    /**
     * Get channel and set init invokeContext if invokeContext not {@code null}
     */
    protected Channel getChannelAndInitInvokeContext(Url url, RequestContext requestContext) throws RemotingException, InterruptedException {
        long start = System.currentTimeMillis();
        long startInNano = System.nanoTime();
        Channel channel;
        try {
            channel = this.channelManager.getAndCreateIfAbsent(url);
        } finally {
            if (null != requestContext) {
                requestContext.putIfAbsent(RequestContext.CLIENT_CONN_CREATETIME, (System.currentTimeMillis() - start));
                requestContext.putIfAbsent(RequestContext.CLIENT_CONN_CREATE_START_IN_NANO, startInNano);
                requestContext.putIfAbsent(RequestContext.CLIENT_CONN_CREATE_END_IN_NANO, System.nanoTime());
            }
        }
        return channel;
    }
}
